#include "dbenv.hxx"
#include "datastore.hxx"
#include <iostream>
#include <sstream>
#include <assert.h>
#include "dbt.hxx"

std::string mymapfunc(Dbt& key) {
    return "mypartition";
}    

int sequentialWrite(DbEnv* env, DataStore* store) {
    DbStatus ret;

    std::string value(128, 'y'); Dbt valuedbt(value); 
    for(int i = 0; i < 100000; ++i)
    {
        std::ostringstream oss; oss<<i; 
        std::string key = oss.str();
        Dbt keydbt(key);
        ret = store->put(keydbt, valuedbt);
        //if(i%1000 == 0) std::cout<<i<<"\n";
        if(!ret.isok())  {std::cout<<"put failed"<<ret.msg(); return -1;}
    }


    return 0;
}

int deleteOdd(DbEnv* env, DataStore* store) {
    DbStatus ret;

    std::string value(128, 'y'); Dbt valuedbt(value);
    for(int i = 1; i < 100000; i +=2)
    {
        std::ostringstream oss; oss<<i;
        std::string key = oss.str(); Dbt keydbt(key);
        ret = store->del(keydbt);
        if(!ret.isok())  {std::cout<<"delete failed"<<ret.msg(); return -1;}
    
        Dbt valuedbt;
        ret = store->get(keydbt, valuedbt); 
         if(ret.getCode() != DbStatus::DB_KEYNOTFOUND)
                             std::cout<<i<<"notdeleted"<<i<<" "<<std::endl;
         
        assert(ret.getCode() == DbStatus::DB_KEYNOTFOUND);
    }

    return 0;
}


int deleteEven(DbEnv* env, DataStore* store) {
    DbStatus ret;

    for(int i = 0; i < 100000; i +=2)
    {
        std::ostringstream oss; oss<<i;
        std::string key = oss.str(); Dbt keydbt(key);
        ret = store->del(keydbt);
        if(!ret.isok())  {std::cout<<"delete failed"<<ret.msg(); return -1;}
    
        Dbt valuedbt;
        ret = store->get(keydbt, valuedbt);
         if(ret.getCode() != DbStatus::DB_KEYNOTFOUND)
                             std::cout<<i<<"notdeleted"<<i<<" "<<std::endl;

        assert(ret.getCode() == DbStatus::DB_KEYNOTFOUND);
    }

    return 0;
}


int sequentialread1(DbEnv* env, DataStore* store) {
    
    for(int i = 0; i < 100000; ++i) {
    
        std::ostringstream oss; oss<<i;
        std::string key = oss.str(); 
        Dbt keydbt(key), valuedbt;
        DbStatus ret = store->get(keydbt, valuedbt);

        if(i%2) {
            if(ret.getCode() != DbStatus::DB_KEYNOTFOUND)
                std::cout<<i<<"notdeleted"<<i<<" "<<std::endl;
            assert(ret.getCode() == DbStatus::DB_KEYNOTFOUND);
        }
        else {

            std::string value0(128, 'y');
            if(value0 != valuedbt.toString()) {
                std::cout<<i<<" not matched: "<<std::endl;
            }
        }
    }
}


int sequentialread2(DbEnv* env, DataStore* store) {

    for(int i = 0; i < 100000; ++i)
    {
        std::ostringstream oss; oss<<i;
        std::string key = oss.str();
        Dbt keydbt(key), valuedbt;
        DbStatus ret = store->get(keydbt, valuedbt);

        if(ret.getCode() != DbStatus::DB_KEYNOTFOUND)
            std::cout<<i<<"notdeleted"<<i<<" "<<std::endl;
        assert(ret.getCode() == DbStatus::DB_KEYNOTFOUND);
    }
}


int main(int argc, char** argv) {

    std::string logmsgFile = argv[1];
    logmsgFile = logmsgFile + "/" + "mydb.msg";

    DbEnv* env = new DbEnv(argv[1], argv[1], logmsgFile, 5*1024*1024);

    uint32_t flag = LOG_HAS_DATA;
    DbStatus ret = env->open(flag);
    if(!ret.isok()) {std::cout<<"env cannot be opened"<<ret.msg(); return -1;}

   DataStore* store = new DataStore(*env);
   const std::string dsName("mystore");
   uint32_t cacheSize = 1000000; 
   uint32_t bufferSize = 10000; 
   uint32_t indexArrayCapacity = 64*1024; 
   uint32_t segCapacity = 32*1024*1024; 
   MAPPINGFUNC* mapFunc = mymapfunc;
   ret = store->open(dsName, cacheSize, indexArrayCapacity, segCapacity,
           mapFunc, 0);
   if(!ret.isok())  {std::cout<<"open"<<ret.msg(); return -1;}

   ret = store->addPartition("mypartition");
   if(!ret.isok())  {std::cout<<"add partition"<<ret.msg(); return -1;}

   ret = env->start(); 
   if(!ret.isok())  {std::cout<<"env start"<<ret.msg(); return -1;}

   std::cout<<"writing mode\n";
   sequentialWrite(env, store);

   std::cout<<"delete odd mode";
   deleteOdd(env, store);

   std::cout<<"reading mode1\n";
   sequentialread1(env, store);

   std::cout<<"delete even mode"; 
   deleteEven(env, store);
   
   std::cout<<"reading mode2\n";  
   sequentialread2(env, store);
 
   env->sync();


   delete store;
   delete env;
}

